package com.hong.spring.config;

import lombok.extern.log4j.Log4j2;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListener;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

/**
 * @author: csh
 * @Date: 2021/3/16 09:43
 * @Description:rocketmq生产者
 */
@Log4j2
public class Consumer {
    //生产实例
    private DefaultMQPushConsumer defaultMQConsumer;
    //生产组
    private String producerGroup;
    //地址
    private String namesrvAddr;
    //topic
    private String topic;
    //tag
    private String tag="*";
    //监听
    private MessageListener messageListener;

    /**
     * 初始化
     */
    public void init() throws MQClientException {
        // 参数信息
        log.info("DefaultMQProducer 初始化");
        log.info(producerGroup);
        log.info(namesrvAddr);
        // 初始化
        defaultMQConsumer = new DefaultMQPushConsumer(producerGroup);
        defaultMQConsumer.setNamesrvAddr(namesrvAddr);
        defaultMQConsumer.setInstanceName(String.valueOf(System.currentTimeMillis()));
        defaultMQConsumer.subscribe(topic,tag);
        // 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
        // 如果非第一次启动，那么按照上次消费的位置继续消费
        defaultMQConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 设置为集群消费(区别于广播消费) CLUSTERING:集群消费  BROADCASTING:广播消费
        defaultMQConsumer.setMessageModel(MessageModel.CLUSTERING);
        defaultMQConsumer.setMessageListener(messageListener);
        defaultMQConsumer.start();
        log.info("rocketmqConsumer start success!");
    }


    public void destroy() {
        defaultMQConsumer.shutdown();
    }


    public void setProducerGroup(String producerGroup) {
        this.producerGroup = producerGroup;
    }

    public void setNamesrvAddr(String namesrvAddr) {
        this.namesrvAddr = namesrvAddr;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public String getTag() {
        return tag;
    }

    public void setTag(String tag) {
        this.tag = tag;
    }

    public void setDefaultMQConsumer(DefaultMQPushConsumer defaultMQConsumer) {
        this.defaultMQConsumer = defaultMQConsumer;
    }

    public void setMessageListener(MessageListener messageListener) {
        this.messageListener = messageListener;
    }
}
